/*
- @Author: Wangjun
- @Date: 2022-11-04 16:11:29

  - @LastEditTime: 2023-12-12 21:24:00

  - @LastEditors: wangjun haodreams@163.com

- @Description:从麦杰取数据写入clickhouse
  - @FilePath: \xr_node_calc\models\boss.go

- hnxr
*/
package models

import (
	"context"
	"strings"

	//"encoding/json"

	"fmt"
	"os"
	"strconv"
	"time"

	"gitee.com/haodreams/godriver/opdb"
	"gitee.com/haodreams/golib/ck"
	"gitee.com/haodreams/golib/exception"
	"gitee.com/haodreams/golib/logs"
	"gitee.com/haodreams/libs/config"
	"gitee.com/haodreams/libs/easy"
	"gitee.com/haodreams/libs/ee"
	"gitee.com/haodreams/libs/routine"
	"gitee.com/haodreams/op_node/kafka"
	"gitee.com/haodreams/op_node/noderdb"
	"gitee.com/haodreams/op_node/thirdparty"
	"gitee.com/hnxr_1/nodex"
	"gitee.com/hnxr_1/nodex/nodetime"
	"github.com/twmb/franz-go/pkg/kgo"
)

type Boss struct {
	memorys  []*Memory
	nodeMems []*nodex.Memory //算子缓存

	produceTopic string //kafka生产数据的topic

	chanKafka     chan *kgo.Record
	chanCk        chan *NodeTask //写入ck
	ckdb          ck.DB          //线上的ck环境,6状态需要使用的
	period        int64
	OpenplantTask //麦杰实时库的任务
}

func (m *Boss) Setup() (err error) {
	noderdb.SetItemer(m)

	err = m.EnvSetup()
	if err != nil {
		return ee.Print(err, logs.CbWarn)
	}

	//初始化ck
	ckDsn := config.String("ck.dsn")
	loadCK := true
	if ckDsn == "" {
		loadCK = false
	}

	m.period = config.DefaultInt64("calc.period", 1000)
	err = m.ckdb.Setup(ckDsn, loadCK)
	ee.Print(err, logs.CbWarn)
	m.ckdb.Exec(sqlCkTask)

	//加载麦杰画布的任务
	err = m.OpenplantTask.Setup()
	ee.Print(err, logs.CbWarn)

	go m.producerKafka()
	go m.write2ck()
	logs.Info("计算周期:", m.period, "毫秒")
	logs.Info("全部任务初始化完成")
	return nil
}

func (m *Boss) EnvSetup() (err error) {
	m.chanKafka = make(chan *kgo.Record, 1000)
	m.chanCk = make(chan *NodeTask, 1000)
	return
}

/**
 * @description: 主线程循环函数
 * @return {*}
 */
func (m *Boss) Loop() {
	defer exception.CatchException()
	var err error
	for routine.IsRunning() {
		now := time.Now()
		unix := now.Unix()
		nodetime.SetTime(unix)
		// unix -= unix % 5

		nowCalc := time.Now()
		err = m.OpenplantTask.Read()
		if err != nil {
			logs.Warn(err)
		}
		m.calc(unix)

		logs.Info("计算耗时:", time.Since(nowCalc))
		m.SaveMemory()

		err = m.OpenplantTask.Write(unix)
		if err != nil {
			logs.Warn(err)
		}

		logs.Info("全链路耗时:", time.Since(now))
		easy.Sleep(&now, m.period)
	}
}

/**
 * @description: 获取任务数据
 * @param {*} path
 * @param {string} category
 * @param {int} devid
 * @return {*}
 */
func (m *Boss) GetTaskData(path, category string, devid int) (data any, err error) {
	path = strings.TrimSuffix(path, ".json")
	ss := strings.Split(path, ".")
	if len(ss) < 2 {
		return
	}

	data = m.OpenplantTask.GetTaskData(ss[0])
	return
}

/**
 * @description: 计算主线程
 * @param {int64} unix
 * @return {*}
 */
func (m *Boss) calc(unix int64) {
	//算子时间计时
	nodetime.SetTime(unix)
	m.OpenplantTask.Run(unix)
}

func (m *Boss) GetOPTask() []*TaskDesc {
	return m.OpenplantTask.tasksDesc
}

/**
 * @description: 获取麦杰实时库测点
 * @param {string} name 测点名
 * @param {bool} isRead 取数方式true=read false=write
 * @return {*}
 */
func (m *Boss) GetOPValue(name string, isRead bool) *opdb.Real {
	return m.OpenplantTask.GetOPValue(name, isRead)
}

// 新增加一个内存计算点，内存点数据不需要外部接口读入也不需要写入到其他地方
func (m *Boss) GetMemItem(category string, id int, name string) *thirdparty.Item {
	return m.OpenplantTask.GetMemItem(name)
}

/**
 * @description: 获取类别的任务列表
 * @param {int} id
 * @param {string} category
 * @return {*}
 */
func (m *Boss) GetTaskList(id int, category string) []*Task {

	return nil
}

/**
 * @description: 生产kafka数据
 * @return {*}
 */
func (m *Boss) producerKafka() {
	m.produceTopic = config.String("kafka.topic.producer")
	if m.produceTopic == "" {
		logs.Info("禁用kafka写入功能")
		return
	}
	for routine.IsRunning() {
		m.producerWork()
		time.Sleep(time.Second * 5)
	}
}

// 生产工作者线程
func (m *Boss) producerWork() {
	defer exception.CatchException()
	writer := kafka.NewWriter()
	max := 3000
	box := easy.NewBox[*kgo.Record](max)

	f := func(key string) {
		now := time.Now()
		err := m.sendKafka(writer, box.Data())
		if err != nil {
			writer.Close()
			time.Sleep(time.Second)
			writer = kafka.NewWriter()
		}
		logs.Info("写入kafka 耗时:", time.Since(now), "写入数据量:", box.Len())
		box.Reset()
	}

	for routine.IsRunning() {
		select {
		case business := <-m.chanKafka:
			if box.IsFull() {
				f(fmt.Sprintf("%s-%d", time.Now().Format(easy.YMDHMS), box.Len()))
			}
			if box.Push(business) {
				f(fmt.Sprintf("%s-%d", time.Now().Format(easy.YMDHMS), box.Len()))
			}
			for len(m.chanKafka) > 0 {
				business = <-m.chanKafka
				if box.Push(business) {
					f(fmt.Sprintf("%s-%d", time.Now().Format(easy.YMDHMS), box.Len()))
				}
			}
		case <-time.After(time.Second):
			if box.Len() > 0 {
				f(fmt.Sprintf("%s-%d", time.Now().Format(easy.YMDHMS), box.Len()))
			}
		}
	}
	//延迟2秒保证数据全部写入kafka
	time.Sleep(time.Second * 2)
	for len(m.chanKafka) > 0 {
		business := <-m.chanKafka
		if box.Push(business) {
			f(fmt.Sprintf("%s-%d", time.Now().Format(easy.YMDHMS), box.Len()))
		}
	}
	if box.Len() > 0 {
		f(fmt.Sprintf("%s-%d", time.Now().Format(easy.YMDHMS), box.Len()))
	}
	writer.Close()
	logs.Info("退出kafka写入功能")
}

func (m *Boss) sendKafka(writer *kgo.Client, msgs []*kgo.Record) (err error) {
	//now := time.Now()
	ctx := context.Background()
	err = writer.ProduceSync(ctx, msgs...).FirstErr()
	if err != nil {
		data, err := json.Marshal(msgs)
		if err != nil {
			logs.Warn(err)
			return err
		}
		path := "data/failed/" + time.Now().Format(easy.YMD)
		os.MkdirAll(path, 0755)
		os.WriteFile(path+"/"+strconv.Itoa(int(time.Now().UnixMicro()))+".json", data, 0644)
		return ee.Print(err, logs.CbWarn)
	}
	return
}

/**
 * @description: 写入ck
 * @return {*}
 */
func (m *Boss) write2ck() {
	if !m.ckdb.Enabled() {
		return
	}
	for routine.IsRunning() {
		m.ckWork()
		time.Sleep(time.Second * 5)
	}
}

// 生产工作者线程
func (m *Boss) ckWork() {
	defer exception.CatchException()
	max := 3000
	box := easy.NewBox[*NodeTask](max)

	f := func(key string) {
		now := time.Now()
		_, err := m.ckdb.InsertObjects(box.Data())
		if err != nil {
			time.Sleep(time.Second)
			_, err = m.ckdb.InsertObjects(box.Data())
			if err != nil {
				//备份没写入成功的
				data, err := json.Marshal(box.Data())
				if err != nil {
					logs.Warn(err.Error())
					return
				}
				path := "data/ck/failed/" + time.Now().Format(easy.YMD)
				os.MkdirAll(path, 0755)
				path = path + "/" + strconv.Itoa(int(time.Now().UnixMicro())) + ".json"
				os.WriteFile(path+".tmp", data, 0644)
				os.Rename(path+".tmp", path)
			}
		}
		logs.Info("写入CK 耗时:", time.Since(now), "写入数据量:", box.Len())
		box.Reset()
	}

	for routine.IsRunning() {
		select {
		case business := <-m.chanCk:
			if box.IsFull() {
				f(fmt.Sprintf("%s-%d", time.Now().Format(easy.YMDHMS), box.Len()))
			}
			if box.Push(business) {
				f(fmt.Sprintf("%s-%d", time.Now().Format(easy.YMDHMS), box.Len()))
			}
			for len(m.chanKafka) > 0 {
				business = <-m.chanCk
				if box.Push(business) {
					f(fmt.Sprintf("%s-%d", time.Now().Format(easy.YMDHMS), box.Len()))
				}
			}
		case <-time.After(time.Second):
			if box.Len() > 0 {
				f(fmt.Sprintf("%s-%d", time.Now().Format(easy.YMDHMS), box.Len()))
			}
		}
	}
	//延迟2秒保证数据全部写入kafka
	time.Sleep(time.Second * 2)
	for len(m.chanKafka) > 0 {
		business := <-m.chanCk
		if box.Push(business) {
			f(fmt.Sprintf("%s-%d", time.Now().Format(easy.YMDHMS), box.Len()))
		}
	}
	if box.Len() > 0 {
		f(fmt.Sprintf("%s-%d", time.Now().Format(easy.YMDHMS), box.Len()))
	}
	logs.Info("退出CK写入功能")
}
